package com.perfma.xtest.cmd;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

public class TriggerCallbackThread {

    private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);

    private static TriggerCallbackThread instance = new TriggerCallbackThread();

    public static TriggerCallbackThread getInstance() {
        return instance;
    }

    static {
        logger.info("执行器回调中心: 启动后台回调线程");
        instance.start();
    }

    /**
     * job results callback queue
     */
    private LinkedBlockingQueue<HandleCallbackParam> callBackQueue = new LinkedBlockingQueue<HandleCallbackParam>();

    public static void pushCallBack(HandleCallbackParam callback) {
        getInstance().callBackQueue.add(callback);
        logger.debug(">>>>>>>>>>> 任务, 任务请求回调, 任务ID:{}", callback.getLogId());
    }

    /**
     * callback thread
     */
    private Thread triggerCallbackThread;
    private volatile boolean toStop = false;

    public void start() {

        triggerCallbackThread = new Thread(new Runnable() {

            @Override
            public void run() {

                // normal callback
                while (!toStop) {
                    try {
                        HandleCallbackParam callback = getInstance().callBackQueue.take();
                        if (callback != null) {

                            // callback list param
                            List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                            int drainToNum = callBackQueue.drainTo(callbackParamList);
                            callbackParamList.add(callback);
                            // callback, will retry if error
                            if (callbackParamList != null && callbackParamList.size() > 0) {
                                for (HandleCallbackParam callbackParam : callbackParamList) {
                                    doCallback(callbackParam);
                                }
                            }
                        }
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
                // last callback
                try {
                    List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                    int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                    if (callbackParamList != null && callbackParamList.size() > 0) {
                        for (HandleCallbackParam handleCallbackParam : callbackParamList) {
                            doCallback(handleCallbackParam);
                        }
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
                logger.warn(">>>>>>>>>>>> 任务, 任务回调线程终止.");

            }
        });
        triggerCallbackThread.setDaemon(true);
        triggerCallbackThread.start();
    }

    public void toStop() {
        toStop = true;
        // interrupt and wait
        triggerCallbackThread.interrupt();
        try {
            triggerCallbackThread.join();
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * do callback, will retry if error
     *
     * @param callbackParam
     */
    private void doCallback(HandleCallbackParam callbackParam) {
        // callback, will retry if error
        try {
            ReturnT<String> callbackResult = callbackParam.getCallBackBiz().callback(callbackParam);
            if (callbackResult != null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
            } else {
                logger.info(">>>>>>>>>>> 任务回调失败, 回调参数为:{}, 回调结果为:{}", new Object[]{callbackParam, callbackResult});
            }
        } catch (Exception e) {
            logger.error(">>>>>>>>>>> 任务回调异常, 回调参数为：{}", callbackParam, e);
        }
    }

}
